package spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.tasks_manager.tasks_manager.node.Node;
import spring.cloud.tasks.tasks_manager.tasks_manager.service.AllotTaskItemService;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.executor.Executor;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.task.TaskItem;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * 域下重排，移除已经存在所有executor，重新获取executors，重新获取作业shards
 */
@Slf4j
@Setter
@Getter
public class TasksFullTaskItemTask extends TaskItemAllotTask {
    public TasksFullTaskItemTask(AllotTaskItemService allotTaskItemService) {
        super(allotTaskItemService);
    }

    @Override
    protected void logStartInfo() {
        log.info("Execute the {} ", this.getClass().getSimpleName());
    }

    @Override
    protected boolean pick(List<String> taskIdList,//
                           List<String> enableTaskIdList,//
                           List<TaskItem> taskItemList,//
                           List<Executor> onlineExecutorList,//
                           List<Executor> onlineOwnTrafficExecutorList) throws Exception {
        // 修正所有executor对所有作业的jobNameList
        for (int j = 0; j < taskIdList.size(); j++) {
            refreshExecutorTaskIdList(onlineExecutorList, taskIdList.get(j));
        }

        // 获取该域下所有enable作业的所有分片
        for (int i = 0; i < enableTaskIdList.size(); i++) {
            String taskId = enableTaskIdList.get(i);
            taskItemList.addAll(createTaskItemList(taskId));
        }

        return true;
    }

    @Override
    protected List<Executor> customLastOnlineExecutorList() throws Exception {
        if (!(curatorFramework.checkExists().forPath(Node.getExecutorsNodePath()) != null)) {
            return new ArrayList<>();
        }
        // 从$SaturnExecutors节点下，获取所有正在运行的Executor
        List<String> executorIdList = curatorFramework.getChildren().forPath(Node.getExecutorsNodePath());
        if (executorIdList == null) {
            return new ArrayList<>();
        }

        List<Executor> lastOnlineExecutorList = new ArrayList<>();
        for (int i = 0; i < executorIdList.size(); i++) {
            String executorId = executorIdList.get(i);
            if (curatorFramework.checkExists().forPath(Node.getExecutorIpNodePath(executorId)) != null) {
                byte[] ip = curatorFramework.getData().forPath(Node.getExecutorIpNodePath(executorId));
                if (ip != null) {
                    Executor executor = new Executor();
                    executor.setExecutorId(executorId);
                    executor.setIp(new String(ip, StandardCharsets.UTF_8));
                    executor.setLostTraffic(getExecutorLostTraffic(executorId));
                    executor.setTaskIdList(new ArrayList<String>());
                    executor.setTaskItemList(new ArrayList<TaskItem>());
                    lastOnlineExecutorList.add(executor);
                }
            }
        }
        return lastOnlineExecutorList;
    }

}
